package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Examine_OneExamine extends BaseSQLApp {
    public static void main(String[] args) {
        new Examine_OneExamine().start(
                1231,
                4,
                Constant.TOPIC_DB
        );
    }

    @Override
    public void handle1(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        readOdsDS(tableEnv,"dwd_exam_exam");

        Table testExamTable = tableEnv.sqlQuery(
                "select " +
                        " `data`['id'] exam_id ," +
                        " `data`['paper_id'] paper_id ," +
                        " `data`['user_id'] user_id ," +
                        " CAST(`data`['score'] as bigint) score ," +
                        " CAST(`data`['duration_sec'] as bigint) duration_sec ," +
                        " ts ," +
                        " et ," +
                        " pt " +
                        " from topic_db where `table`='test_exam' and `data`['deleted'] = '0' " +
                        " and (`type`='insert' or `type`='update')"
        );
        //testExamTable.execute().print();
        tableEnv.createTemporaryView("test_exam_table",testExamTable);

        Table paperTable = tableEnv.sqlQuery(
                "select `data`['id'] paper_id ," +
                        " `data`['paper_title'] paper_title ," +
                        " `data`['course_id'] course_id " +
                        " from topic_db where `table`='test_paper' and `data`['deleted'] = '0' " +
                        " and (`type`='insert' or `type`='update') "
        );
        //paperTable.execute().print();
        tableEnv.createTemporaryView("test_paper_table",paperTable);

        Table dwdExamExamTable = tableEnv.sqlQuery(
                "select t1.exam_id , t1.paper_id , t2.paper_title , t1.user_id , t1.score , t1.duration_sec , t2.course_id , ts " +
                        " from test_exam_table as t1 " +
                        " join test_paper_table as t2 " +
                        " on t1.paper_id = t2.paper_id "
        );


        tableEnv.executeSql(
                "create table dwd_exam_exam (" +
                        " exam_id string ," +
                        " paper_id string ," +
                        " paper_title string ," +
                        " user_id string ," +
                        " score bigint ," +
                        " duration_sec bigint ," +
                        " course_id string ," +
                        " ts bigint ," +
                        " PRIMARY KEY (exam_id) NOT ENFORCED " +
                        " ) " + SQLUtil.getUpsertKafkaDDL("dwd_exam_exam")
        );

        dwdExamExamTable.executeInsert("dwd_exam_exam");


    }
/*
{"before":null,"after":
    {"source_table":"test_exam","source_type":"insert","sink_table":"dwd_examination_test_paper",
    "sink_columns":"id,paper_id,user_id,score,duration_sec,create_time,submit_time,update_time"},
"source":
    {"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,
    "snapshot":"false","db":"edu_config","sequence":null,"table":"table_process_dwd","server_id":0,
    "gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},
    "op":"r","ts_ms":1731568972726,"transaction":null}
*/

        }

